package org.agile4j.rpc.server;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;

import org.agile4j.rpc.common.bean.RpcRequest;
import org.agile4j.rpc.common.bean.RpcResponse;
import org.agile4j.rpc.common.codec.RpcDecoder;
import org.agile4j.rpc.common.codec.RpcEncoder;
import org.agile4j.rpc.registry.ServiceRegistry;
import org.agile4j.rpc.server.annotation.RpcService;
import org.apache.commons.collections4.MapUtils;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;

import java.util.HashMap;
import java.util.Map;

/**
 * 实现 RPC 服务器
 * 使用 Netty 可实现一个支持 NIO 的 RPC 服务器，
 * 需要使用ServiceRegistry注册服务地址
 * Created by Administrator on 2016/8/12.
 */
public class RpcServer implements ApplicationContextAware,InitializingBean{
    private static final Logger LOGGER = LoggerFactory.getLogger(RpcServer.class);

    private String serverAddress ;
    private ServiceRegistry serviceRegistry ;

    /**
     * 存放 服务名 与 服务对象 之间的映射关系
     */
    private Map<String, Object> handlerMap = new HashMap<>() ;

    public RpcServer(String serverAddress) {
        this.serverAddress = serverAddress;
    }

    public RpcServer(String serverAddress, ServiceRegistry serviceRegistry) {
        this.serverAddress = serverAddress;
        this.serviceRegistry = serviceRegistry;
    }

    public void setApplicationContext(ApplicationContext ctx) throws BeansException {
        // 扫描带有 RpcService 注解的类并初始化 handlerMap 对象
        Map<String, Object> serviceBeanMap = ctx.getBeansWithAnnotation(RpcService.class) ;
        if (MapUtils.isNotEmpty(serviceBeanMap)) {
            for (Object serviceBean : serviceBeanMap.values()) {
                RpcService rpcService = serviceBean.getClass().getAnnotation(RpcService.class) ;
                String serviceName = rpcService.value().getName() ;
                String serviceVersion = rpcService.version() ;
                if (StringUtils.isNotEmpty(serviceVersion)) {
                    serviceName += "-" + serviceVersion ;
                }
                handlerMap.put(serviceName, serviceBean) ;
            }
        }
    }

    @Override
    public void afterPropertiesSet() throws Exception {
        EventLoopGroup bossGroup = new NioEventLoopGroup() ;
        EventLoopGroup workerGroup = new NioEventLoopGroup() ;
        try {
            // 创建并初始化 Netty 服务端 Bootstrap 对象
            ServerBootstrap bootstrap = new ServerBootstrap() ;
            bootstrap.group(bossGroup, workerGroup).channel(NioServerSocketChannel.class) ;
            bootstrap.childHandler(new ChannelInitializer<SocketChannel>() {
                @Override
                protected void initChannel(SocketChannel channel) throws Exception {
                    ChannelPipeline pipeline = channel.pipeline();
                    pipeline.addLast(new RpcDecoder(RpcRequest.class));// 解码 RPC 请求（为了处理请求）
                    pipeline.addLast(new RpcEncoder(RpcResponse.class));// 编码 RPC 响应（为了返回响应）
                    pipeline.addLast(new RpcServerHandler(handlerMap));// 处理 RPC 请求
                }
            });
            /**
             * BACKLOG用于构造服务端套接字ServerSocket对象，标识当服务器请求处理线程全满时，
             * 用于临时存放已完成三次握手的请求的队列的最大长度。
             * 如果未设置或所设置的值小于1，Java将使用默认值50。
             */
            bootstrap.option(ChannelOption.SO_BACKLOG, 1024) ;
            /**
             * 是否启用心跳保活机制。在双方TCP套接字建立连接后（即都进入ESTABLISHED状态）
             * 并且在两个小时左右上层没有任何数据传输的情况下，这套机制才会被激活
             */
            bootstrap.childOption(ChannelOption.SO_KEEPALIVE, true);

            // 获取 RPC 服务器的 IP 地址与端口号
            String[] addressArray = StringUtils.split(serverAddress, ":") ;
            String ip = addressArray[0] ;
            int port = Integer.parseInt(addressArray[1]) ;
            // 启动 RPC 服务器
            ChannelFuture future = bootstrap.bind(ip, port).sync();
            // 注册 RPC 服务地址
            if (serviceRegistry != null) {
                for (String interfaceName : handlerMap.keySet()) {
                    serviceRegistry.register(interfaceName, serverAddress);
                    LOGGER.debug("registry service: {} => {}", interfaceName, serverAddress);
                }
            }
            LOGGER.debug("server started on port {}", port);
            // 关闭 RPC 服务器
            future.channel().closeFuture().sync();
        } finally {
            workerGroup.shutdownGracefully();
            bossGroup.shutdownGracefully() ;
        }
    }
}
